package cn.dfun.sample.flink.apitest
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * process function示例
  */
object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val inputStream = env.socketTextStream("node-01", 7777)

    // 包装成样例类
    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
//      .keyBy(_.id)
//      .process(new MyKeyedProcessFunction)

    // 需求: 10s内温度连续上升则报警
    // 窗口模式只要不是连续统计总会有覆盖不到的情况
    val warningStream = dataStream
        .keyBy(_.id)
        .process(new TempIncreWarning(10000))
//        .timeWindow(Time.seconds(10))
    warningStream.print()

    env.execute("process function test")
  }
}

class TempIncreWarning(interval: Long) extends KeyedProcessFunction[String, SensorReading, String] {
  // 保存上一个温度值进行比较,保存注册定时器的时间戳用于删除
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTempState", classOf[Double]))
  lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timerTsState", classOf[Long]))

  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
    var lastTemp = lastTempState.value()
    var timeTs = timerTsState.value()
    // 更新温度值
    lastTempState.update(i.temperature)
    // 判断当前温度值和上次温度比较,如果温度上升且没有定时器
    if(i.temperature > lastTemp && timeTs == 0) {
      val ts = context.timerService().currentProcessingTime() + interval
      context.timerService().registerProcessingTimeTimer(ts)
      timerTsState.update(ts)
      // 温度下降删除定时器
    } else if(i.temperature < lastTemp) {
      context.timerService().deleteProcessingTimeTimer(timeTs)
      // 清空状态
      timerTsState.clear()
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect("传感器" + ctx.getCurrentKey + "的温度连续" + interval/1000 + "秒连续上升")
    timerTsState.clear()
  }
}

// KeyedProcessFunction功能测试
class MyKeyedProcessFunction extends KeyedProcessFunction[String, SensorReading, String] {
  var myState: ValueState[Int] = _

  override def open(parameters: Configuration): Unit = {
//    myState = new ValueState[Int](new ValueStateDescriptor[Int]("myState", classOf[Int]))
  }
  // 可以实现filter
  override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
    // 获取当前key
    context.getCurrentKey
    // 获取时间戳
    context.timestamp()
    // 获取watermark
    context.timerService().currentWatermark()
    // 注册定时器
    context.timerService().registerEventTimeTimer(context.timestamp() + 60000L)
//    context.timerService().deleteEventTimeTimer()
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {

  }
}
